package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.BeanToJsonStrFunction;
import com.atguigu.gmall.realtime.app.func.DimAsyncFunction;
import com.atguigu.gmall.realtime.beans.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.utils.DateFormatUtil;
import com.atguigu.gmall.realtime.utils.DorisUtil;
import com.atguigu.gmall.realtime.utils.KafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

/**
 * @author Felix
 * @date 2023/8/10
 * 下单聚合统计
 *      维度：sku
 *      度量：原始金额、实付金额、优惠券减免金额、活动减免金额
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、redis、doris、DwdTradeOrderDetail、DwsTradeSkuOrderWindow
 * 开发步骤
 *      -环境准备
 *      -检查点相关的设置
 *      -从kafka主题中读取数据
 *      -过滤空消息并对读取到的数据进行类型转换    jsonStr->jsonObj
 *      -去重
 *          *为什么会产生空消息以及重复数据
 *              我们是从dwd的下单事实表中读取数据，下单事实表由订单明细表、订单表、订单明细活动表、订单明细优惠券表
 *              4张表组成，在和明细活动以及明细优惠券进行关联的时候，使用的是左外连接，如果左外连接的时候左表数据先到
 *              右表数据后到的话，会产生如下3条数据
 *              [左表     null] +I
 *              [左表     null] -D
 *              [左表     右表] +I
 *              这样的数据发送到kafka的下单事实表主题中，kafka主题会接收到3条消息
 *              左表  null
 *              null
 *              左表  右表
 *         *去重前需要先按照唯一(订单明细id)键进行分组
 *         *去重方式1：状态 + 定时器
 *              第一条数据到来的时候，先不往下游发送，放到状态中缓存起来，并注册一个5s之后执行的定时器
 *              如果第二条重复数据到来的话，会用第二条数据的聚合时间和状态中缓存的数据的聚合时间进行比较，
 *              将时间大的放到状态中
 *              当定时器被触发执行的时候，将状态中的数据发送到下游
 *              缺点：时效性差
 *         *去重方式2：状态 + 抵消
 *              第一条数据来的时候，直接将数据发送到下游，并在状态中进行缓存
 *              如果第二条重复数据到来了，会将状态中缓存的数据影响到度量值的属性进行取反，发送到下游
 *              再讲第二条数据发下游发送
 *      -再次对流中数据进行类型转换  jsonObj->实体类对象
 *      -指定Watermark的生成策略以及提取事件时间字段
 *      -按照统计的维度sku进行分组
 *      -开窗
 *      -聚合计算
 *      ~~~~~~~~~~~~~~~~~~~华丽的分界线~~~~~~~~~~~~~~~~~~~~~~
 *      -维度关联
 *          *基本维度关联的实现
 *              HbaseUtil-JSONObject getDimInfoFromHbase()
 *          *优化1：旁路缓存
 *              思路：先从个缓存中查询维度数据，如果在缓存中找到了要关联的维度，直接将查询到维度数据作为方法的返回值
 *                   进行返回(缓存命中)；如果在缓存中没有找到要关联的维度，发送请求到Hbase中查询维度数据，并将查询
 *                   到的维度放到缓存中缓存起来，方便下次查询使用
 *              缓存产品选型
 *                  redis：  性能不错，维护性好   √
 *                  状态：    性能很好，维护性差
 *              Redis相关的设置
 *                  key:    维度表名:主键值
 *                  type:   String
 *                  expire: 1day
 *                  注意：如果维度数据发生了变化，清除Redis中缓存的数据
 *              DimUtil -- JSONObject getDimInfo
 *          *优化2：异步IO
 *              为什么要使用异步IO？
 *                  在处理流中数据，进行维度关联的时候，如果使用map算子，默认使用的同步的处理方式，处理完一个元素
 *                  再处理下一个元素，如果和外部系统进行交互的，性能较低。
 *                  如果单纯想提升处理能力，也可以加大并行度，但是更大的并行度，意味着需要更多的硬件资源，不可能
 *                  无限制的加大，所以在硬件资源有限的情况下，要想提升单个并行度的处理能力，可以使用异步IO
 *              Flink提供了发送异步请求的API
 *                  AsyncDataStream.[un]orderedWait(
 *                      流,
 *                      如果发送异步请求,
 *                      60,
 *                      时间单位
 *                  )
 *             单独封装了一个类，发送异步请求进行维度关联
 *                  class DimAsyncFunction extends RichAsyncFunction implements DimJoinFunction{
 *                      open:
 *                          初始化异步Redis连接
 *                          初始化异步hbase连接
 *                      close:
 *                          关闭异步Redis连接
 *                          关闭异步hbase连接
 *                      asyncInvoke：发送异步请求进行维度关联
 *                          CompletableFuture.supplyAsync{
 *                              开始执行CompletableFuture任务，支持返回值
 *                              【发送异步请求到Redis中获取维度数据】
 *                          }.thenApplyAsync{
 *                              第一个任务执行完成后，执行第二个回调方法任务，会将第一个任务的执行结果，作为入参
 *                              ，传递到第二个任务回调方法中，并且回调方法是有返回值的
 *                              【根据上一个任务返回结果，判断是否要发送异步请求到hbase中获取维度数据】
 *                          }.thenAcceptAsync{
 *                              第一个任务执行完成后，执行第二个回调方法任务，会将第一个任务的执行结果，作为入参
 *                              ，传递到第二个任务回调方法中，但是回调方法是没有返回值的
 *                              【将查询出的维度属性补充到流中对象上，并发送到下游】
 *                          }
 *                          abstract String getKey(obj);
 *                          abstract join(obj,dimJsonObj)
 *                  }
 *              模板方法设计模式
 *                  在父类中定义完成某一个功能的核心算法的骨架(步骤)，但是某些步骤需要延迟到子类中去实现
 *                  在不改变父类核心算法估计的前提下，每一个子类都可以有自己不同的实现。
 *      -将结果写到Doris
 */
public class DwsTradeSkuOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关设置
        env.enableCheckpointing(5000L);
        //TODO 3.从kafka主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_sku_order_group";
        //3.2 创建消费者对象
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topic, groupId);
        //3.3 消费数据  封装为流
        DataStreamSource<String> kafkaStrDS
            = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        //TODO 4.过滤空消息 并对流中数据类型进行转换 jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
            new ProcessFunction<String, JSONObject>() {
                @Override
                public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                    if (StringUtils.isNotEmpty(jsonStr)) {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        out.collect(jsonObj);
                    }
                }
            }
        );
        // {"create_time":"2023-08-08 09:39:10","sku_num":"1","activity_rule_id":"5",
        // "split_original_amount":"11999.0000","split_coupon_amount":"0.0","sku_id":"19",
        // "date_id":"2023-08-08","user_id":"2031","province_id":"30","activity_id":"4",
        // "sku_name":"液晶平板电视机","id":"14258663","order_id":"55563","split_activity_amount":"1199.9",
        // "split_total_amount":"10799.1","ts":"1691631550"}
        // jsonObjDS.print(">>>>");

        //TODO 5.去重
        //5.1 按照唯一键(订单明细的id)进行分组
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS
            = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));
        /*//5.2 去重方式1 状态 + 定时器
        //注意：这种方式每条数据处理的时候都需要花费5s的时间，对整个实时计算的时效性影响较大
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                private ValueState<JSONObject> lastValueState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<JSONObject> valueStateDescriptor
                        = new ValueStateDescriptor<JSONObject>("lastValueState", JSONObject.class);
                    lastValueState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    //从状态中获取数据
                    JSONObject lastValueJsonObj = lastValueState.value();
                    if (lastValueJsonObj == null) {
                        //说明不重复  将当前这条数据放到状态中
                        lastValueState.update(jsonObj);
                        //注册一个5s之后执行的定时器
                        long currentProcessingTime = ctx.timerService().currentProcessingTime();
                        ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                    } else {
                        //说明重复  在当前这条数据来之前，已经有相同的订单明细id数据过来了
                        //获取状态中存放的数据的聚合时间
                        String ts1 = lastValueJsonObj.getString("聚合时间");
                        //获取当前处理的数据的聚合时间
                        String ts2 = jsonObj.getString("聚合时间");
                        if (ts2.compareTo(ts1) >= 0) {
                            lastValueState.update(jsonObj);
                        }
                    }
                }

                @Override
                public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                    //从状态中获取数据
                    JSONObject jsonObj = lastValueState.value();
                    //将状态中的对象传递到下游
                    out.collect(jsonObj);
                    //清状态
                    lastValueState.clear();
                }
            }
        );*/
        //5.3 去重方式2 状态 + 抵消
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                private ValueState<JSONObject> lastValueState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<JSONObject> valueStateDescriptor
                        = new ValueStateDescriptor<JSONObject>("lastValueState", JSONObject.class);
                    valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10)).build());
                    lastValueState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    //从状态中获取数据
                    JSONObject lastValeJsonObj = lastValueState.value();
                    if(lastValeJsonObj != null){
                        //说明产生重复   对当前数据中影响到度量值的字段进行一个取反操作(抵消)
                        String splitOriginalAmount = lastValeJsonObj.getString("split_original_amount");
                        String splitCouponAmount = lastValeJsonObj.getString("split_coupon_amount");
                        String splitActivityAmount = lastValeJsonObj.getString("split_activity_amount");
                        String splitTotalAmount = lastValeJsonObj.getString("split_total_amount");
                        lastValeJsonObj.put("split_original_amount","-" + splitOriginalAmount);
                        lastValeJsonObj.put("split_coupon_amount","-" + splitCouponAmount);
                        lastValeJsonObj.put("split_activity_amount","-" + splitActivityAmount);
                        lastValeJsonObj.put("split_total_amount","-" + splitTotalAmount);
                        out.collect(lastValeJsonObj);
                    }
                    out.collect(jsonObj);
                    lastValueState.update(jsonObj);
                }
            }
        );
        // distinctDS.print(">>>");
        //TODO 6.再次对流中数据进行类型转换      jsonObj -> 实体类对象
        SingleOutputStreamOperator<TradeSkuOrderBean> beanDS = distinctDS.map(
            new MapFunction<JSONObject, TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean map(JSONObject jsonObj) throws Exception {
                    // {"create_time":"2023-08-08 09:39:10","sku_num":"1","activity_rule_id":"5",
                    // "split_original_amount":"11999.0000","split_coupon_amount":"0.0","sku_id":"19",
                    // "date_id":"2023-08-08","user_id":"2031","province_id":"30","activity_id":"4",
                    // "sku_name":"液晶平板电视机","id":"14258663","order_id":"55563","split_activity_amount":"1199.9",
                    // "split_total_amount":"10799.1","ts":"1691631550"}
                    String skuId = jsonObj.getString("sku_id");
                    String splitOriginalAmount = jsonObj.getString("split_original_amount");
                    String splitCouponAmount = jsonObj.getString("split_coupon_amount");
                    String splitActivityAmount = jsonObj.getString("split_activity_amount");
                    String splitTotalAmount = jsonObj.getString("split_total_amount");
                    Long ts = jsonObj.getLong("ts") * 1000;
                    TradeSkuOrderBean orderBean = TradeSkuOrderBean.builder()
                        .skuId(skuId)
                        .originalAmount(new BigDecimal(splitOriginalAmount))
                        .couponAmount(new BigDecimal(splitCouponAmount))
                        .activityAmount(new BigDecimal(splitActivityAmount))
                        .orderAmount(new BigDecimal(splitTotalAmount))
                        .ts(ts)
                        .build();
                    return orderBean;
                }
            }
        );
        // beanDS.print(">>>");
        //TODO 7.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<TradeSkuOrderBean> withWatermarkDS = beanDS.assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<TradeSkuOrderBean>forMonotonousTimestamps()
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<TradeSkuOrderBean>() {
                        @Override
                        public long extractTimestamp(TradeSkuOrderBean orderBean, long recordTimestamp) {
                            return orderBean.getTs();
                        }
                    }
                )
        );

        //TODO 8.按照统计维度sku进行分组
        KeyedStream<TradeSkuOrderBean, String> skuIdKeyedDS
            = withWatermarkDS.keyBy(TradeSkuOrderBean::getSkuId);
        //TODO 9.开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS
            = skuIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        //TODO 10.聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
            new ReduceFunction<TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                    value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                    value1.setCouponAmount(value1.getCouponAmount().add(value2.getCouponAmount()));
                    value1.setActivityAmount(value1.getActivityAmount().add(value2.getActivityAmount()));
                    value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                    return value1;
                }
            },
            new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                @Override
                public void apply(String s, TimeWindow window, Iterable<TradeSkuOrderBean> input, Collector<TradeSkuOrderBean> out) throws Exception {
                    String stt = DateFormatUtil.toYmdHms(window.getStart());
                    String edt = DateFormatUtil.toYmdHms(window.getEnd());
                    String curDate = DateFormatUtil.toPartitionDate(window.getStart());
                    for (TradeSkuOrderBean orderBean : input) {
                        orderBean.setStt(stt);
                        orderBean.setEdt(edt);
                        orderBean.setCurDate(curDate);
                        out.collect(orderBean);
                    }
                }
            }
        );
        // reduceDS.print(">>>");
        //TODO 11.关联sku维度
        /*SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
            new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                private Connection hbaseConn;
                private Jedis jedis;

                @Override
                public void open(Configuration parameters) throws Exception {
                    hbaseConn = HbaseUtil.getHbaseConnection();
                    jedis = RedisUtil.getRedisClient();
                }

                @Override
                public void close() throws Exception {
                    HbaseUtil.closeHbaseConnection(hbaseConn);
                    RedisUtil.closeRedisClient(jedis);
                }

                @Override
                public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                    //根据流中对象获取要关联的维度的主键
                    String skuId = orderBean.getSkuId();
                    //根据维度的主键获取对应的维度对象
                    JSONObject dimInfoJsonObj = DimUtil.getDimInfo(
                        jedis, hbaseConn, GmallConfig.HBASE_NAMESPACE, "dim_sku_info", skuId
                    );
                    //将维度对象的相关维度属性补充到流中的对象上
                    if(dimInfoJsonObj != null){
                        //id,spu_id,price,sku_name,sku_desc,weight,tm_id,category3_id,sku_default_img,is_sale,create_time
                        orderBean.setSkuName(dimInfoJsonObj.getString("sku_name"));
                        orderBean.setSpuId(dimInfoJsonObj.getString("spu_id"));
                        orderBean.setTrademarkId(dimInfoJsonObj.getString("tm_id"));
                        orderBean.setCategory3Id(dimInfoJsonObj.getString("category3_id"));
                    }

                    return orderBean;
                }
            }
        );
        // withSkuInfoDS.print(">>>");*/
        /*SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
            new DimMapFunction<TradeSkuOrderBean>("dim_sku_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimInfoJsonObj) {
                    orderBean.setSkuName(dimInfoJsonObj.getString("sku_name"));
                    orderBean.setSpuId(dimInfoJsonObj.getString("spu_id"));
                    orderBean.setTrademarkId(dimInfoJsonObj.getString("tm_id"));
                    orderBean.setCategory3Id(dimInfoJsonObj.getString("category3_id"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSkuId();
                }
            }
        );
        */
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
            reduceDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_sku_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimInfoJsonObj) {
                    orderBean.setSkuName(dimInfoJsonObj.getString("sku_name"));
                    orderBean.setSpuId(dimInfoJsonObj.getString("spu_id"));
                    orderBean.setTrademarkId(dimInfoJsonObj.getString("tm_id"));
                    orderBean.setCategory3Id(dimInfoJsonObj.getString("category3_id"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSkuId();
                }
            },
            60,
            TimeUnit.SECONDS
        );
        // withSkuInfoDS.print(">>>");
        //TODO 12.关联spu维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withSpuInfoDS = AsyncDataStream.unorderedWait(
            withSkuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_spu_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimInfoJsonObj) {
                    orderBean.setSpuName(dimInfoJsonObj.getString("spu_name"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSpuId();
                }
            },
            60,
            TimeUnit.SECONDS
        );
        //TODO 13.关联tm维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withTmDS = AsyncDataStream.unorderedWait(
            withSpuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_trademark") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimInfoJsonObj) {
                    orderBean.setTrademarkName(dimInfoJsonObj.getString("tm_name"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getTrademarkId();
                }
            },
            60,
            TimeUnit.SECONDS
        );
        // withTmDS.print(">>>");
        //TODO 14.关联category3维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c3Stream = AsyncDataStream.unorderedWait(
            withTmDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category3") {

                @Override
                public String getKey(TradeSkuOrderBean bean) {
                    return bean.getCategory3Id();
                }

                @Override
                public void join(TradeSkuOrderBean bean,
                                 JSONObject dim) {
                    bean.setCategory3Name(dim.getString("name"));
                    bean.setCategory2Id(dim.getString("category2_id"));
                }
            },
            60,
            TimeUnit.SECONDS
        );

        //TODO 15.关联category2维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c2Stream = AsyncDataStream.unorderedWait(
            c3Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category2") {

                @Override
                public String getKey(TradeSkuOrderBean bean) {
                    return bean.getCategory2Id();
                }

                @Override
                public void join(TradeSkuOrderBean bean,
                                 JSONObject dim) {
                    bean.setCategory2Name(dim.getString("name"));
                    bean.setCategory1Id(dim.getString("category1_id"));
                }
            },
            60,
            TimeUnit.SECONDS
        );
        //TODO 16.关联category1维度
        SingleOutputStreamOperator<TradeSkuOrderBean> restulStream = AsyncDataStream.unorderedWait(
            c2Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category1") {

                @Override
                public String getKey(TradeSkuOrderBean bean) {
                    return bean.getCategory1Id();
                }

                @Override
                public void join(TradeSkuOrderBean bean,
                                 JSONObject dim) {
                    bean.setCategory1Name(dim.getString("name"));
                }
            },
            60,
            TimeUnit.SECONDS
        );

        restulStream.print(">>>");

        //TODO 17.将关联的结果写到Doris
        restulStream.map(
            new BeanToJsonStrFunction<>()
        ).sinkTo(DorisUtil.getDorisSink("dws_trade_sku_order_window"));
        env.execute();
    }

}